package com.patsnap.data.npd.dw.etl.job.ods.dispatcher;

import com.patsnap.data.npd.dw.etl.job.ods.serialization.PreOdsInfo;
import com.patsnap.data.npd.dw.repository.DBOperator;
import org.reflections.Reflections;

import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

public class SinkRouterDispatcher<I> implements Serializable {
    private final Map<String, Class<? extends LaunchProcessor>> launchProcessorClassMap = new HashMap<>();
    private final Map<String, LaunchProcessor> launchProcessorMap = new ConcurrentHashMap<>();

    // 扫描这个包下面所有有注解的类
    private static final String SCAN_PACKAGE_NAME = "com.patsnap.data.npd.dw.etl.job.ods.parser";
    private static final String PREFIX_KEY = "%s_%s_%s";
    private final DataLayer dataLayer;

    private static final String ETL_TYPE_NEWS = "npd_news";
    private static final String ETL_TYPE_JOURNAL = "npd_journal";
    private static final String ETL_TYPE_PAPER = "npd_paper";

    public SinkRouterDispatcher(DataLayer dataLayer) {
        this.dataLayer = dataLayer;

        Reflections reflections = new Reflections(SCAN_PACKAGE_NAME);
        Set<Class<? extends OdsInfoLaunchProcessor>> processClasses = reflections.getSubTypesOf(OdsInfoLaunchProcessor.class);
        processClasses.stream().filter(f -> f.isAnnotationPresent(ProcessorIdentifier.class)).forEach(processClass -> {
            ProcessorIdentifier processorInstance = processClass.getAnnotation(ProcessorIdentifier.class);
            String parser = processorInstance.parser();
            String type = processorInstance.type();
            // type -> npd_journal
            // parser -> JOURNAL_PARSER_BEST
            // dispatchKey -> ODS_npd_journal_JOURNAL_PARSER_BEST
            String dispatchKey = String.format(PREFIX_KEY, dataLayer, type, parser);
            launchProcessorClassMap.put(dispatchKey, processClass);
        });
    }

    /**
     * Dispatch data by different operations for every record.
     * Initialize target instance and call the launch methods to execute business flow
     *
     * @param record input record
     */
    public void dispatcher(I record, Map<String, Object> globalConfig, DBOperator dbOperator) {
        if (record instanceof PreOdsInfo) {
            PreOdsInfo preOdsInfo = (PreOdsInfo) record;
            String dispatchKey = String.format(PREFIX_KEY, dataLayer, preOdsInfo.getType(), preOdsInfo.getParser());
            ((OdsInfoLaunchProcessor) getLaunchProcessor(dispatchKey, globalConfig, dbOperator)).launchWithJsonRecord(preOdsInfo);
        }
    }

    /**
     * Initialize target instance
     *
     * @param dispatchKey located destination
     * @return new processor instance
     */
    private LaunchProcessor getLaunchProcessor(String dispatchKey, Map<String, Object> globalConfig, DBOperator dbOperator) {
        return launchProcessorMap.computeIfAbsent(dispatchKey, f -> {
            try {
                return launchProcessorClassMap.get(f).getConstructor(Map.class, DBOperator.class).newInstance(globalConfig, dbOperator);
            } catch (Exception e) {
                throw new RuntimeException("Initialize Processor failed! dispatchKey => " + dispatchKey, e);
            }
        });
    }
}
